package com.swak.boot.config.zookeeper;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.client.ZKClientConfig;

import com.swak.zookeeper.ZookeeperService;
import com.swak.zookeeper.curator.CuratorZookeeperService;

public class ZookeeperBuilder {

	private static ZookeeperService ME = null;

	private static void configZookeeper() {
		System.setProperty(ZKClientConfig.ENABLE_CLIENT_SASL_KEY, "false");
	}

	public static ZookeeperService build(ZookeeperProperties properties) throws InterruptedException {
		if (ME != null) {
			return ME;
		}
		synchronized (CuratorZookeeperService.class) {
			if (ME != null) {
				return ME;
			}
			configZookeeper();
			CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
					.connectString(properties.getAddress()).retryPolicy(new RetryNTimes(1, 1000))
					.connectionTimeoutMs(properties.getConnectionTimeoutMs())
					.sessionTimeoutMs(properties.getSessionTimeoutMs());
			String authority = properties.getAuthority();
			if (authority != null && authority.length() > 0) {
				builder = builder.authorization("digest", authority.getBytes());
			}
			CuratorFramework client = builder.build();
			CuratorZookeeperService me = new CuratorZookeeperService(client, properties.getConnectionTimeoutMs());
			ME = me;
		}
		return ME;
	}
}
